package cn.itcast.streaming.window.function;

import cn.itcast.streaming.entity.ItcastDataObj;
import cn.itcast.streaming.entity.TripModel;
import com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;

/**
 * Author itcast
 * Date 2021/9/24 15:11
 * 行驶里程根据不同车速统计soc，里程，油耗，开始时间，结束时间，累计里程等
 * 开发步骤：
 * //1.重写 apply 方法
 * //1.1 将迭代器转换成集合列表
 * //1.2 对集合列表的数据进行排序操作
 * //1.3 将集合对象转换成 TripModel 对象返回
 * //1.4 将 TripModel 对象收集返回
 * //2.将驾驶行程指标计算getTripModel，得到 TripModel 对象
 * //2.1 获取第一条数据
 * //2.2 将 vin（车架号） ，tripStartTime（行程开始时间），start_BMS_SOC（行程开始Soc），start_longitude（行程开始经度），start_latitude（行程开始维度），start_mileage（行程开始表显里程数）
 * 传递给 tripModel 对象。
 * //2.3 从最后一条数据中得到对象并赋值给 tripEndTime（行程结束时间），end_BMS_SOC（行程结束soc），end_longitude（行程结束经度），end_latitude（行程结束维度），end_mileage（行程结束表显里程数），mileage（行程驾驶公里数），time_comsuption（行程消耗时间）、这里存储的是分钟数，lastSoc（上次的行程Soc）、将当前行程开始的电量消耗百分比作为上一个行程结束的电量消耗百分比，lastMileage（上次的里程数）
 * //2.4 遍历 itcastDataObj list获取一下内容
 * // 每条数据的速度
 * //获取上次行程报文的soc
 * //计算每条数据的soc与lastSoc进行比较差值 socDiff
 * //如果socDiff大于0，那么赋值给 soc_comsuption
 * //如果speed大于tripModel对象中保存的最大车速并且小于150，将speed保存为最大速度
 * //低速行驶 speed >=0 && <40,低速行驶个数+1；low_BMS_SOC=low_BMS_SOC+(当次油耗和上次油耗之差)；setLow_BMS_Mileage=setLow_BMS_Mileage+（当次里程和上次里程之差）
 * //中速行驶 >=40 && <80
 * //高速行驶 >80 && <150
 * //设置当前的soc作为 lastSoc 上次的soc 和将当次的里程作为 lastMileage 上次的里程
 * //增加扩展字段，判断当前是否有异常数据，如果列表长度大于1说明是正常行程0，否则是异常行程1
 */
public class DriveTripWindowFunction extends RichWindowFunction<ItcastDataObj, TripModel, String, TimeWindow> {

    //1.重写 apply 方法
    @Override
    public void apply(String key, TimeWindow window, Iterable<ItcastDataObj> input, Collector<TripModel> out) throws Exception {
        //1.1 将迭代器转换成集合列表
        ArrayList<ItcastDataObj> itcastDataArr = Lists.newArrayList(input);
        //1.2 对集合列表的数据进行排序操作
        Collections.sort(itcastDataArr, Comparator.comparingLong(ItcastDataObj::getTerminalTimeStamp));
        //1.3 将集合对象转换成 TripModel 对象返回
        TripModel tripModel = getTripModel(itcastDataArr);
        //1.4 将 TripModel 对象收集返回
        out.collect(tripModel);

    }

    /**
     * 将15分钟窗口内的所有元素封装成TripModel对象
     *
     * @param itcastDataArr
     * @return TripModel
     */
    //2.将驾驶行程指标计算getTripModel，得到 TripModel 对象
    private TripModel getTripModel(ArrayList<ItcastDataObj> itcastDataArr) {
        TripModel tripModel = new TripModel();
        //2.1 获取第一条数据
        ItcastDataObj firstItcastData = itcastDataArr.get(0);
        //2.2 将 vin（车架号） ，tripStartTime（行程开始时间），start_BMS_SOC（行程开始Soc），
        // start_longitude（行程开始经度），start_latitude（行程开始维度），
        // start_mileage（行程开始表显里程数） 传递给 tripModel 对象。
        tripModel.setVin(firstItcastData.getVin());
        tripModel.setTripStartTime(firstItcastData.getTerminalTime());
        tripModel.setStart_BMS_SOC(firstItcastData.getSoc());
        tripModel.setStart_longitude(firstItcastData.getLng());
        tripModel.setStart_latitude(firstItcastData.getLat());
        tripModel.setStart_mileage(firstItcastData.getTotalOdometer());
        //2.3 从最后一条数据中得到对象并赋值给 tripEndTime（行程结束时间），end_BMS_SOC（行程结束soc），
        ItcastDataObj lastItcastDataObj = itcastDataArr.get(itcastDataArr.size() - 1);
        // end_longitude（行程结束经度），end_latitude（行程结束维度），end_mileage（行程结束表显里程数），
        tripModel.setTripEndTime(lastItcastDataObj.getTerminalTime());
        tripModel.setEnd_BMS_SOC(lastItcastDataObj.getSoc());
        tripModel.setEnd_longitude(lastItcastDataObj.getLng());
        tripModel.setEnd_latitude(lastItcastDataObj.getLat());
        tripModel.setEnd_mileage(lastItcastDataObj.getTotalOdometer());
        // mileage（行程驾驶公里数），
        tripModel.setMileage(lastItcastDataObj.getTotalOdometer() - firstItcastData.getTotalOdometer());
        // time_comsuption（行程消耗时间）、这里存储的是分钟数，lastSoc（上次的行程Soc）、
        tripModel
                .setTime_comsuption((lastItcastDataObj.getTerminalTimeStamp() - firstItcastData.getTerminalTimeStamp()) / 1000 / 60D);
        tripModel.setLastSoc(firstItcastData.getSoc() + 0D);
        // 将当前行程开始的电量消耗百分比作为上一个行程结束的电量消耗百分比，lastMileage（上次的里程数）
        tripModel.setLastMileage(firstItcastData.getTotalOdometer());
        //2.4 遍历 itcastDataObj list获取一下内容
        for (ItcastDataObj itcastDataObj : itcastDataArr) {
            // 每条数据的速度
            Double currentSpeed = itcastDataObj.getSpeed();
            //获取上次行程报文的soc
            Double lastSoc = tripModel.getLastSoc();
            //计算每条数据的soc与lastSoc进行比较差值 socDiff
            int currentSoc = itcastDataObj.getSoc();
            double socDiff = Math.abs(currentSoc - lastSoc);
            //如果socDiff大于0，那么赋值给 soc_comsuption
            double soc_comsuption = 0d;
            if (socDiff > 0) {
                soc_comsuption = socDiff;
            }
            //如果speed大于tripModel对象中保存的最大车速并且小于150，将speed保存为最大速度
            if(currentSpeed > tripModel.getMax_speed()){
                tripModel.setMax_speed(currentSpeed);
            }
            //低速行驶 speed >=0 && <40,低速行驶个数+1；low_BMS_SOC=low_BMS_SOC+(当次油耗和上次油耗之差)；
            // setLow_BMS_Mileage=setLow_BMS_Mileage+（当次里程和上次里程之差）
            if(currentSpeed >=0 && currentSpeed<40){
                tripModel.setTotal_low_speed_nums(tripModel.getTotal_low_speed_nums()+1);
                tripModel.setLow_BMS_Mileage(Math.abs(itcastDataObj.getTotalOdometer()-tripModel.getLastMileage()));
                tripModel.setLow_BMS_SOC(soc_comsuption);
                //中速行驶 >=40 && <80
            }else if(currentSpeed>=40 && currentSpeed<80){
                tripModel.setTotal_medium_speed_nums(tripModel.getTotal_medium_speed_nums()+1);
                tripModel.setMedium_BMS_Mileage(Math.abs(itcastDataObj.getTotalOdometer()-tripModel.getLastMileage()));
                tripModel.setMedium_BMS_SOC(soc_comsuption);
                //高速行驶 >80 && <150
            }else if(currentSpeed>=80 && currentSpeed<150){
                tripModel.setTotal_high_speed_nums(tripModel.getTotal_high_speed_nums()+1);
                tripModel.setHigh_BMS_Mileage(Math.abs(itcastDataObj.getTotalOdometer()-tripModel.getLastMileage()));
                tripModel.setHigh_BMS_SOC(soc_comsuption);
            }
            //设置当前的soc作为 lastSoc 上次的soc 和将当次的里程作为 lastMileage 上次的里程
            tripModel.setLastSoc(currentSoc+0d);
            tripModel.setLastMileage(itcastDataObj.getTotalOdometer()+0d);
            //增加扩展字段，判断当前是否有异常数据，如果列表长度大于1说明是正常行程0，否则是异常行程1
            if(itcastDataArr.size()>0){
                tripModel.setTripStatus(0);
            }else{
                tripModel.setTripStatus(1);
            }
        }
        return tripModel;
    }
}
